package user.service.Impl;

import com.spring4all.spring.boot.starter.hbase.api.HbaseTemplate;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang.time.DateFormatUtils;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.Get;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.util.Bytes;
import org.springframework.stereotype.Service;
import user.contstant.Constants;
import user.service.IHBasePassService;
import user.util.RowKeyGenUtil;
import user.vo.PassTemplate;

import java.io.IOException;

/**
 * pass HBase 服务
 * 消费者消费优惠券
 */
@Service
@Slf4j
public class HBasePassServiceImpl implements IHBasePassService {

    /** HBase 客户端*/
    private final HbaseTemplate hbaseTemplate;

    public HBasePassServiceImpl(HbaseTemplate hbaseTemplate) {
        this.hbaseTemplate = hbaseTemplate;
    }

    /**
     * 写入HBase
     * @param passTemplate{@link PassTemplate}
     * @return true/false
     */
    @Override
    public boolean dropPassTemplateToHBase(PassTemplate passTemplate) {

        //判断passTemplate是否为空
        if(passTemplate == null)
            return false;
        //获取passTemplate rowKey
        String rowKey = RowKeyGenUtil.genPassTemplateRowKey(passTemplate);
        try{
            /**
             * 与HBase建立连接
             * 将表名转换为TableName对象
             * 判断是否存在rowKey,相同优惠券
             */
            if(hbaseTemplate.getConnection().getTable(TableName.valueOf(Constants.PassTemplateTable.TABLE_NAME))
                    .exists(new Get(Bytes.toBytes(rowKey)))){
                log.warn("RowKey {} is already exist!",rowKey);
                return false;
            }
        } catch (IOException e) {
            log.error("dropPassTemplateToHBase error :{}", e.getMessage());
            return false;
        }
        //写入HBase
        Put put = new Put(Bytes.toBytes(rowKey));
        //列族B信息
        put.addColumn(Bytes.toBytes(Constants.PassTemplateTable.FAMILY_B),Bytes.toBytes(Constants.PassTemplateTable.ID),
                      Bytes.toBytes(passTemplate.getId()));
        put.addColumn(Bytes.toBytes(Constants.PassTemplateTable.FAMILY_B),
                      Bytes.toBytes(Constants.PassTemplateTable.TITLE),
                      Bytes.toBytes(passTemplate.getTitle()));
        put.addColumn(Bytes.toBytes(Constants.PassTemplateTable.FAMILY_B),
                      Bytes.toBytes(Constants.PassTemplateTable.SUMMARY),
                      Bytes.toBytes(passTemplate.getSummary()));
        put.addColumn(Bytes.toBytes(Constants.PassTemplateTable.FAMILY_B),
                Bytes.toBytes(Constants.PassTemplateTable.DESC),
                Bytes.toBytes(passTemplate.getDesc()));
        put.addColumn(Bytes.toBytes(Constants.PassTemplateTable.FAMILY_B),
                Bytes.toBytes(Constants.PassTemplateTable.HAS_TOKEN),
                Bytes.toBytes(passTemplate.getHasToken()));
        put.addColumn(Bytes.toBytes(Constants.PassTemplateTable.FAMILY_B),
                Bytes.toBytes(Constants.PassTemplateTable.BACKGROUND),
                Bytes.toBytes(passTemplate.getBackground()));

        //列族c信息
        put.addColumn(Bytes.toBytes(Constants.PassTemplateTable.FAMILY_C),
                Bytes.toBytes(Constants.PassTemplateTable.LIMIT),
                Bytes.toBytes(passTemplate.getLimit()));
        put.addColumn(Bytes.toBytes(Constants.PassTemplateTable.FAMILY_C),
                Bytes.toBytes(Constants.PassTemplateTable.START),
                Bytes.toBytes(DateFormatUtils.ISO_DATE_FORMAT.format(passTemplate.getStart())));
        put.addColumn(Bytes.toBytes(Constants.PassTemplateTable.FAMILY_C),
                Bytes.toBytes(Constants.PassTemplateTable.END),
                Bytes.toBytes(DateFormatUtils.ISO_DATE_FORMAT.format(passTemplate.getEnd())));

        //写入HBase
        hbaseTemplate.saveOrUpdate(Constants.PassTemplateTable.TABLE_NAME,put);
        return true;
    }
}
